package com.sqk.dxpro.dxtop;

import java.io.IOException;


import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.TreeMap;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;

import com.sqk.dxpro.utils.TProperties;

public class DxTopReducer extends Reducer<Text,Text,NullWritable,Text>{

	
	public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 
		
		Text result = new Text();
		
		int topvalue = Integer.parseInt(context.getConfiguration().get("topvalue"));
		int topn = Integer.parseInt(context.getConfiguration().get("topn"));
		TreeMap<Integer, String> topMap = new TreeMap<Integer, String>();
		
	    for (Text val : values) {
		String[] str = val.toString().split(TProperties.getValue("fileoutsplit"));
		//数据放入集合
		topMap.put(Integer.parseInt(str[topvalue]), val.toString());
			if(topMap.size() > topn){
				topMap.remove(topMap.firstKey());//移除超过取值条数的数据
			}
	    }
	    int num = 0;
	    Iterator<Integer> iterator = topMap.keySet().iterator();
	    //将得到的数组转换，用来排名
	    List<String> list = new ArrayList<String>();
	    while (iterator.hasNext()) {
		list.add(topMap.get(iterator.next()));
	    }
	    //将数据的调数+1可以计算排名
	    num = list.size() + 1 ;
	    for(int i = 0 ;i< list.size() ; i++){
		//得出排名
		num = num -1 ;
		//行为ID，用户ID，PV（访问次数），排名
		result = new Text(list.get(i) + TProperties.getValue("outfilesplit") + num );
			context.write(NullWritable.get(), result);
	    }
	}

}
